package paas.storage.distributedColumnDatabase.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson.parser.Feature;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
import org.springframework.stereotype.Service;
import paas.storage.common.config.ConnectionManager;
import paas.storage.common.config.HBaseDaoUtil;
import paas.storage.common.constant.ResponseCode;
import paas.storage.common.utils.AssertUtils;
import paas.storage.distributedColumnDatabase.IData;
import paas.storage.dto.DataQueryConditionsParam;
import paas.storage.dto.DataQueryVo;
import paas.storage.dto.InsertModel;
import paas.storage.dto.KeyValueVo;
import paas.storage.exception.RRException;
import paas.storage.result.*;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 备注
 *
 * @author xufeng
 * @email 525207937@qq.com
 * @date 2021/1/25 14:31
 */
@Service
public class IDataImpl implements IData {

    private static Logger log = LoggerFactory.getLogger(IDataImpl.class);

    @Autowired
    private ConnectionManager connectionManager;
    @Autowired
    private HBaseDaoUtil hBaseDaoUtil;

    /**
     * 插入数据
     * [{row_key,columnfamily:column,value},{}]。
     *
     * @param connectionId
     * @param database         数据库
     * @param table            必填
     * @param columnFamilyList 必填  列簇列表 使用JSON方式定义的列簇值列表，定义方式：
     *                         [{
     *                         "row_key": "row1",
     *                         "columnfamily": "column",
     *                         "value": "45"
     *                         }, {
     *                         "rowKey": "row2",
     *                         "columnfamily": "cf2:c1",
     *                         "value": 12
     *                         }]
     * @param expendParams     以json字符串格式定义：
     *                         {key1:value1, key2:value2,e 。
     * @return
     */
    @Override
    public DataInsertResponse insert(String connectionId, String database, String table, String columnFamilyList, String expendParams) {
        DataInsertResponse dataInsertResponse = new DataInsertResponse();
        dataInsertResponse.setTaskStatus(ResponseCode.TASK_STATUS_OK);
        try {
            AssertUtils.isTrue(StringUtils.isEmpty(connectionId), "连接不能为空");
            if (database != null) {
                AssertUtils.charLengthLe(database, 64, "连接长度为128以下");
            }
            if (table != null) {
                AssertUtils.charLengthLe(table, 64, "表名不能为空");
            }
            AssertUtils.isTrue(StringUtils.isEmpty(columnFamilyList), "列簇不能为空");


            if (connectionManager.getConnection(connectionId) == null) {
                dataInsertResponse.setTaskStatus(0);
                dataInsertResponse.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                dataInsertResponse.setErrorMsg("Please establish a connection first");
                return dataInsertResponse;
            }

            //转换bean
            List<InsertModel> insertModels = JSON.parseArray(columnFamilyList, InsertModel.class);

            //表对象
            TableName tn = hBaseDaoUtil.getTable(database, table);

            //新增数据对象
            List<Put> puts = new ArrayList<>();


            Table insertTable = connectionManager.getConnection(connectionId).getTable(tn);
            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            //表不存在
            if (!admin.isTableAvailable(tn)) {
                throw new RRException("表不存在");
            }
            insertModels.forEach(insertModel -> {
                Put put = new Put(Bytes.toBytes(insertModel.getRow_key()));
                String[] columnfamily = insertModel.getColumnfamily().split(":");
                put.addColumn(Bytes.toBytes(columnfamily[0].trim()), Bytes.toBytes(columnfamily[1].trim()), Bytes.toBytes(insertModel.getValue()));
                //pdf上 没有要求插入多列
                puts.add(put);
            });
            insertTable.put(puts);
            insertTable.close();
            admin.close();
        } catch (RRException e) {
            dataInsertResponse.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            dataInsertResponse.setErrorCode(ResponseCode.BUSINESS_CODE);
            dataInsertResponse.setErrorMsg("业务异常," + e.getMessage());
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            dataInsertResponse.setTaskStatus(1);
            dataInsertResponse.setErrorCode(ResponseCode.DATA_INSERT_ERROR);
            dataInsertResponse.setErrorMsg("插入数据失败," + e.getMessage());
        }
        return dataInsertResponse;
    }

    /**
     * @param connectionId
     * @param database         数据库 不必填
     * @param table            必填
     * @param rowKey
     * @param columnFamilyList 列簇信息  "["cf1:c1","cf2"]"
     * @param expendParams
     * @return
     */
    @Override
    public Response delete(String connectionId, String database, String table, String rowKey, String columnFamilyList, String expendParams) {
        Response response = new Response();
        response.setTaskStatus(ResponseCode.TASK_STATUS_OK);
        try {
            if (database != null) {
                AssertUtils.charLengthLe(database, 64, "连接长度为128以下");
            }
            if (table != null) {
                AssertUtils.charLengthLe(table, 64, "表名不能为空");
            }


            if (connectionManager.getConnection(connectionId) == null) {
                response.setTaskStatus(0);
                response.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                response.setErrorMsg("Please establish a connection first");
                return response;
            }
            //表对象
            TableName tn = hBaseDaoUtil.getTable(database, table);

            Table deleteTable = connectionManager.getConnection(connectionId).getTable(tn);
            List<Delete> deletes = new ArrayList<>();
            //列簇不等于空
            if (StringUtils.isNotEmpty(columnFamilyList)) {
                List<String> stringList = JSON.parseArray(columnFamilyList, String.class);
                stringList.forEach(obj -> {
                    String[] columnFamily = obj.split(":");
                    //删除最新版本
                    Delete delete = new Delete(Bytes.toBytes(rowKey));
                    delete.addColumn(Bytes.toBytes(columnFamily[0].trim()), Bytes.toBytes(columnFamily[1].trim()));
                    deletes.add(delete);
                });
            } else {
                deletes.add(new Delete(Bytes.toBytes(rowKey)));
            }
            //根据rowKey删除
            deleteTable.delete(deletes);
            deleteTable.close();
        } catch (RRException e) {
            response.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            response.setErrorCode(ResponseCode.BUSINESS_CODE);
            response.setErrorMsg("业务异常," + e.getMessage());
        } catch (Exception e) {
            response.setTaskStatus(0);
            response.setErrorCode(ResponseCode.DATA_DELETE_ERROR);
            response.setErrorMsg("删除数据库失败," + e.getMessage());
            log.error(e.getMessage(), e);
        }

        return response;
    }

    /**
     * 更新数据
     *
     * @param connectionId
     * @param database         数据库 不必填
     * @param table            表 必填
     * @param columnFamilyList 列簇信息
     * @param expendParams     扩展字段
     *                         [{"row_key":value, "columnfamily":value, "column":value, "value":value_new,
     *                         "version":value}
     * @return
     */
    @Override
    public Response update(String connectionId, String database, String table, String columnFamilyList, String expendParams) {
        Response response = new Response();
        response.setTaskStatus(ResponseCode.TASK_STATUS_OK);
        try {
            if (database != null) {
                AssertUtils.charLengthLe(database, 64, "连接长度为128以下");
            }
            if (table != null) {
                AssertUtils.charLengthLe(table, 64, "表名不能为空");
            }

            if (connectionManager.getConnection(connectionId) == null) {
                response.setTaskStatus(0);
                response.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                response.setErrorMsg("Please establish a connection first");
                return response;
            }

            //转换bean
            List<InsertModel> insertModels = JSON.parseArray(columnFamilyList, InsertModel.class);

            //表对象
            TableName tn = hBaseDaoUtil.getTable(database, table);


            //新增数据对象
            List<Put> puts = new ArrayList<>();


            Table insertTable = connectionManager.getConnection(connectionId).getTable(tn);
            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            //表不存在
            if (!admin.isTableAvailable(tn)) {
                throw new RRException("表不存在");
            }

            insertModels.forEach(insertModel -> {
                //时间戳就是版本号 如果没存就用默认的时间戳
                long version = System.currentTimeMillis();
                if (insertModel.getVersion() != null) {
                    version = insertModel.getVersion();
                }
                Put put = new Put(Bytes.toBytes(insertModel.getRow_key()));
                put.addColumn(Bytes.toBytes(insertModel.getColumnfamily()), Bytes.toBytes(insertModel.getColumn()), version, Bytes.toBytes(insertModel.getValue()));
                puts.add(put);
            });
            insertTable.put(puts);
            admin.close();
            insertTable.close();
        }catch (RRException e) {
            response.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            response.setErrorCode(ResponseCode.BUSINESS_CODE);
            response.setErrorMsg("业务异常," + e.getMessage());
        } catch (IOException e) {
            response.setTaskStatus(0);
            response.setErrorCode(ResponseCode.DATA_UPDATE_ERROR);
            response.setErrorMsg("更新数据失败!," + e.getMessage());
            log.error(e.getMessage(), e);
        }

        return response;
    }

    /**
     * 查询数据
     *
     * @param connectionId
     * @param database         数据库
     * @param table            表
     * @param conditions       必填  查询条件使用JSON方式定义的查询条件列表,支持正则表达式，支持过滤器等用户自定义查询条件， 例如：
     *                         {"MaxVersions":3, "TimeRange": "[454546,677357]","filter":{"SingleColumnValue Filter":{"family": Bytes.
     *                         .toBytes("columnfamily1"),"qualifier": Bytes.toBytes("column1"),"compareOp": CompareOp.EQUAL, "value": Bytes.
     *                         .toBytes("2013-06-08")}}
     * @param startKey         起始键 不必填
     * @param endKey           结束键 不必填
     * @param columnFamilyList 列簇列表 不必填 可传递列族中某个列或整个列族，定义方式：
     *                         ["columnfamily1:column1","columnfamil2:colum n2",:]
     * @param expendParams     以json字符串格式定义：
     *                         {key1:value1, key2:value2,By
     * @return
     */
    @Override
    public DataQueryResponse query(String connectionId, String database, String table, String conditions, String startKey, String endKey, String columnFamilyList, String expendParams) {
        DataQueryResponse response = new DataQueryResponse();
        response.setTaskStatus(ResponseCode.TASK_STATUS_OK);
        try {
            if (database != null) {
                AssertUtils.charLengthLe(database, 64, "连接长度为128以下");
            }
            if (table != null) {
                AssertUtils.charLengthLe(table, 64, "表名长度超出");
            }
            AssertUtils.isTrue(StringUtils.isEmpty(table), "表不能为空");

            AssertUtils.isTrue(StringUtils.isEmpty(conditions), "查询条件不能为空");

            if (connectionManager.getConnection(connectionId) == null) {
                response.setTaskStatus(0);
                response.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                response.setErrorMsg("Please establish a connection first");
                return response;
            }
            //表对象
            TableName tn = hBaseDaoUtil.getTable(database, table);

            DataQueryConditionsParam dataQueryConditionsParam = JSON.parseObject(conditions, DataQueryConditionsParam.class);


            Table queryTable = connectionManager.getConnection(connectionId).getTable(tn);
            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            //表不存在
            if (!admin.isTableAvailable(tn)) {
                throw new RRException("表不存在");
            }
            Scan scan = new Scan();
            if (StringUtils.isNotEmpty(startKey)) {
                scan.setStartRow(Bytes.toBytes(startKey.trim()));
            }
            if (StringUtils.isNotEmpty(endKey)) {
                scan.setStopRow(Bytes.toBytes(endKey.trim()));
            }
            if (dataQueryConditionsParam != null && dataQueryConditionsParam.getMaxVersions() > 0) {
                scan.readVersions(dataQueryConditionsParam.getMaxVersions());
            }
            if (dataQueryConditionsParam != null && StringUtils.isNotEmpty(dataQueryConditionsParam.getTimeRange())) {
                List<Long> time = new ArrayList<>();
                String timeStr = StringUtils.strip(dataQueryConditionsParam.getTimeRange(), "[]");
                String[] times = timeStr.split(",");
                Arrays.asList(times).forEach(s -> time.add(Long.valueOf(s)));
                scan.setTimeRange(time.get(0), time.get(1));
            }
            if (dataQueryConditionsParam != null && dataQueryConditionsParam.getFilter() != null) {
                //最外层对象
                LinkedHashMap<String, Object> map = JSON.parseObject(conditions, new TypeReference<LinkedHashMap<String, Object>>() {
                }, Feature.OrderedField);
                LinkedHashMap<String, Object> param = JSON.parseObject(map.get("filter").toString(), new TypeReference<LinkedHashMap<String, Object>>() {
                }, Feature.OrderedField);
                FilterList filterList = scanFilter(param);
                /*FilterList filterList = new FilterList();
                SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("user".getBytes(),"age".getBytes(),CompareOperator.EQUAL,"20".getBytes());
                filterList.addFilter(singleColumnValueFilter);*/
                if (null != filterList) {
                    scan.setFilter(filterList);
                }
            }
            //扫描某一列
            if (StringUtils.isNotEmpty(columnFamilyList)) {
                List<String> stringList = JSON.parseArray(columnFamilyList, String.class);
                stringList.forEach(obj -> {
                    String[] columnFamily = obj.split(":");
                    scan.addColumn(Bytes.toBytes(columnFamily[0].trim()), Bytes.toBytes(columnFamily[1].trim()));
                });
            }
            ResultScanner scanner = queryTable.getScanner(scan);
            Iterator<Result> resultIterator = scanner.iterator();


            // 进行迭代，拿到Result对象
            List<DataQueryVo> dataQueryVos = new ArrayList<>();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                List<DataQueryVo> dataQueryVo = showResult(result);
                dataQueryVos.addAll(dataQueryVo);
            }
            response.setResult(JSON.toJSONString(dataQueryVos));
            scanner.close();
            admin.close();
            queryTable.close();
        } catch (RRException e) {
            response.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            response.setErrorCode(ResponseCode.BUSINESS_CODE);
            response.setErrorMsg("业务异常," + e.getMessage());
        }catch (Exception e) {
            response.setTaskStatus(0);
            response.setErrorCode(ResponseCode.DATA_QUERY_ERROR);
            response.setErrorMsg("查询数据失败!," + e.getMessage());
            log.error(e.getMessage(), e);

        }

        return response;
    }

    /**
     * 基于键值的数据查询
     *
     * @param connectionId     必填 是
     * @param database         不必填
     * @param table            必填
     * @param rowKey           必填
     * @param columnFamilyList 列簇 不必填 可传递列族中某个列或整个列族，定义方
     *                         式：
     *                         [columnfamily1:column1,c]
     * @param expendParams     扩展参数 以json字符串格式定义：
     *                         {key1:value1, key2:value2,nf。
     * @return
     */
    @Override
    public DataqueryByKeyResponse queryByKey(String connectionId, String database, String table, String rowKey, String columnFamilyList, String expendParams) {
        DataqueryByKeyResponse dataqueryByKeyResponse = new DataqueryByKeyResponse();
        dataqueryByKeyResponse.setTaskStatus(ResponseCode.TASK_STATUS_OK);
        try {
            //表对象
            TableName tn = hBaseDaoUtil.getTable(database, table);
            Table queryTable = connectionManager.getConnection(connectionId).getTable(tn);

            if (connectionManager.getConnection(connectionId) == null) {
                dataqueryByKeyResponse.setTaskStatus(0);
                dataqueryByKeyResponse.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                dataqueryByKeyResponse.setErrorMsg("Please establish a connection first");
                return dataqueryByKeyResponse;
            }


            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            //表不存在
            if (!admin.isTableAvailable(tn)) {
                throw new RRException("表不存在");
            }
            Get get = new Get(Bytes.toBytes(rowKey));


            //Scan scan = new Scan();
            //scan.setStartRow(Bytes.toBytes(rowKey));

            List<String> columnfamily = JSON.parseArray(columnFamilyList, String.class);
            if (columnfamily != null && columnfamily.size() > 0) {
                columnfamily.forEach(s -> {
                    String[] o = s.split(":");
                    //scan.addColumn(Bytes.toBytes(o[0]), Bytes.toBytes(o[1]));
                    get.addColumn(Bytes.toBytes(o[0].trim()), Bytes.toBytes(o[1].trim()));

                });
            }

            //ResultScanner scanner = queryTable.getScanner(scan);
            // Iterator<Result> resultIterator = scanner.iterator();
            Result result = queryTable.get(get);
            dataqueryByKeyResponse.setIteration(result);
            queryTable.close();
        }catch (RRException e) {
            dataqueryByKeyResponse.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            dataqueryByKeyResponse.setErrorCode(ResponseCode.BUSINESS_CODE);
            dataqueryByKeyResponse.setErrorMsg("业务异常," + e.getMessage());
        } catch (Exception e) {
            dataqueryByKeyResponse.setTaskStatus(0);
            dataqueryByKeyResponse.setErrorCode(ResponseCode.DATA_QUERY_KEY_ERROR);
            dataqueryByKeyResponse.setErrorMsg("基于键值查询数据失败!," + e.getMessage());
            log.error(e.getMessage(), e);

        }
        return dataqueryByKeyResponse;
    }

    /**
     * 获取最新的数据
     *
     * @param result
     * @return
     */
    public static List<DataQueryVo> showResult(Result result) {
        List<DataQueryVo> dataQueryVos = new ArrayList<>();
        // 通过Result对象获取行键
        byte[] row = result.getRow();
        // 获取Map对象
        NavigableMap<byte[],
                NavigableMap<byte[],
                        NavigableMap<Long, byte[]>>>
                map = result.getMap();
        // map中的键是什么？   -->     列族
        // map中的值是什么？   -->     某个列族对应的值
        // 遍历map
        Set<byte[]> cfs = map.keySet();
        for (byte[] cf : cfs) {
            NavigableMap<byte[],
                    NavigableMap<Long, byte[]>>
                    map1 = map.get(cf);
            // map1中的键是什么？  -->   列名
            // map1中的值是什么？  -->   某个列对应的值
            // 遍历map1
            Set<byte[]> cols = map1.keySet();
            for (byte[] col : cols) {
                NavigableMap<Long, byte[]>
                        map2 = map1.get(col);
                // map2中的键是什么？  -->   Timestamp
                // map2中的值是什么？  -->   值
                // 遍历map2
                Set<Long> tss = map2.keySet();
                for (Long ts : tss) {
                    byte[] val = map2.get(ts);
                    DataQueryVo dataQueryVo = new DataQueryVo();

                    dataQueryVo.setRow_key(new String(row));
                    dataQueryVo.setColumnfamily(new String(cf));
                    dataQueryVo.setColumn(new String(col));
                    dataQueryVo.setValue(new String(val));
                    dataQueryVo.setVersion(String.valueOf(ts));
                    dataQueryVos.add(dataQueryVo);
                    /*log.info(
                            "行键：" + new String(row) +
                                    "\t列族：" + new String(cf) +
                                    "\t列名：" + new String(col) +
                                    "\t时间戳：" + ts +
                                    "\t值：" + new String(val));*/
                }
            }
        }
        return dataQueryVos;
    }


    /*    *//**
     * scan获取多版本数据
     *//*
    public List<DataQueryVo> showVersionResult(Result result) {
        List<DataQueryVo> dataQueryVos = new ArrayList<>();
        List<Cell> calls = result.listCells();
        //Cell[] cells = result.rawCells();
        if (calls != null) {
            calls.parallelStream().forEachOrdered(cell -> {
                DataQueryVo dataQueryVo = new DataQueryVo();
                dataQueryVo.setRow_key(new String(CellUtil.cloneRow(cell)));
                dataQueryVo.setColumnfamily(new String(CellUtil.cloneFamily(cell)));
                dataQueryVo.setColumn(new String(CellUtil.cloneQualifier(cell)));
                dataQueryVo.setValue(new String(CellUtil.cloneValue(cell)));
                dataQueryVo.setVersion(String.valueOf(cell.getTimestamp()));
                dataQueryVos.add(dataQueryVo);
            });
        }
        return dataQueryVos;
    }*/

    /**
     * @param connectionId 连接id 不能为空
     * @param database     数据库 能为空
     * @param table        表名 不能为空
     * @param filter       过滤器对象 不能为空
     * @param expendParams 扩展参数可能为空
     * @return
     */
    @Override
    public DataQueryByRangeResponse queryByRange(String connectionId, String database, String table, String filter, String expendParams) {
        DataQueryByRangeResponse dataQueryByRangeResponse = new DataQueryByRangeResponse();
        dataQueryByRangeResponse.setTaskStatus(ResponseCode.TASK_STATUS_OK);

        try {
            if (connectionManager.getConnection(connectionId) == null) {
                dataQueryByRangeResponse.setTaskStatus(0);
                dataQueryByRangeResponse.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                dataQueryByRangeResponse.setErrorMsg("Please establish a connection first");
                return dataQueryByRangeResponse;
            }
            TableName tn = hBaseDaoUtil.getTable(database, table);


            Table queryTable = connectionManager.getConnection(connectionId).getTable(tn);
            //表不存在
       /*     if (!admin.isTableAvailable(tn)) {
                throw new RRException("表不存在");
            }*/

            Scan scan = new Scan();

            //最外层对象
            LinkedHashMap<String, Object> map = JSON.parseObject(filter, new TypeReference<LinkedHashMap<String, Object>>() {
            }, Feature.OrderedField);
            FilterList filterList = scanFilter(map);
            if (null != filterList) {
                scan.setFilter(filterList);
            }
            ResultScanner scanner = queryTable.getScanner(scan);
            Iterator<Result> resultIterator = scanner.iterator();
            dataQueryByRangeResponse.setIteration(resultIterator);
            queryTable.close();
        }catch (RRException e) {
            dataQueryByRangeResponse.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            dataQueryByRangeResponse.setErrorCode(ResponseCode.BUSINESS_CODE);
            dataQueryByRangeResponse.setErrorMsg("业务异常," + e.getMessage());
        }  catch (Exception e) {
            dataQueryByRangeResponse.setTaskStatus(0);
            dataQueryByRangeResponse.setErrorCode(ResponseCode.TABLE_QUERY_RANGE_ERROR);
            dataQueryByRangeResponse.setErrorMsg("范围查询失败," + e.getMessage());
            log.error(e.getMessage(), e);

        }

        return dataQueryByRangeResponse;
    }


    public FilterList scanFilter(LinkedHashMap<String, Object> map) {

        //满足所有条件
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        //遍历获取键值对
        map.forEach((s, o) -> {
            Class<?> clazz = null;
            try {
                String[] names = s.split("@");
                // 因为json传参不支持同名key 那么传参相同的约束一下规则
                //SingleColumnValueFilter:{},SingleColumnValueFilter@1
                clazz = Class.forName("org.apache.hadoop.hbase.filter." + names[0]);
            } catch (ClassNotFoundException ignored) {
                throw new RRException("找不到这个查询拦截器");
            }
            //获取这个json的过滤器属性
            LinkedHashMap<String, Object> field = JSON.parseObject(o.toString(), new TypeReference<LinkedHashMap<String, Object>>() {
            }, Feature.OrderedField);

            //参数key
            List<String> strs = new ArrayList<>();
            field.forEach((s1, o2) -> strs.add(s1));
            //参数值
            List<String> row = new ArrayList<>();
            field.forEach((s1, o2) -> row.add(o2.toString()));

            Class<?> finalClazz = clazz;
            // ASM字节码
            LocalVariableTableParameterNameDiscoverer localVariableTableParameterNameDiscoverer = new LocalVariableTableParameterNameDiscoverer();

            AtomicReference<Filter> filterObj = new AtomicReference<>();
            //筛选过滤器
            Arrays.stream(finalClazz.getConstructors()).filter(constructor -> {
                if (constructor.getParameterCount() != field.size()) {
                    return false;
                }
                //拿到构造函数参数名称
                String[] parameterNames = localVariableTableParameterNameDiscoverer.getParameterNames(constructor);
                //判断构造函数后参数名称是否存在. 全部存在返回true
                boolean flag = Arrays.stream(parameterNames).allMatch(strs::contains);
                if (flag) {

                    try {
                        AtomicInteger j = new AtomicInteger();
                        List<KeyValueVo> list = new ArrayList<>();
                        Arrays.stream(constructor.getParameterTypes()).forEachOrdered(aClass -> {
                            KeyValueVo keyValueVo = new KeyValueVo();
                            keyValueVo.setKey(aClass.getName());
                            keyValueVo.setValue(row.get(j.get()));
                            j.getAndIncrement();
                            list.add(keyValueVo);
                        });
                        //最终构造的参数
                        List<Object> objects = new ArrayList<>();
                        //构造函数参数转换
                        list.forEach(keyValueVo -> {

                            if ("long".equals(keyValueVo.getKey())) {
                                objects.add(Long.parseLong(keyValueVo.getValue()));
                            }
                            if ("int".equals(keyValueVo.getKey())) {
                                objects.add(Integer.parseInt(keyValueVo.getValue()));
                            }
                            if ("String".equals(keyValueVo.getKey())) {
                                objects.add(Boolean.parseBoolean(keyValueVo.getValue()));
                            }
                            if ("java.util.List".equals(keyValueVo.getKey())) {
                                String str = keyValueVo.getValue().replaceAll("\\[", "").replaceAll("\\]", "");
                                String[] strings = str.split(",");
                                List<Long> longs = new ArrayList<>();
                                Arrays.asList(strings).forEach(s3 -> {
                                    longs.add(Long.valueOf(s3));
                                });
                                objects.add(longs);

                            }
                            if ("boolean".equals(keyValueVo.getKey())) {
                                objects.add(Boolean.parseBoolean(keyValueVo.getValue()));
                            }
                            if ("[B".equals(keyValueVo.getKey())) {
                                objects.add(Bytes.toBytes(keyValueVo.getValue()));
                            }
                            if ("org.apache.hadoop.hbase.filter.CompareFilter$CompareOp".equals(keyValueVo.getKey())) {
                                CompareFilter.CompareOp compareOp = CompareFilter.CompareOp.valueOf(keyValueVo.getValue());
                                objects.add(compareOp);
                            }

                            if ("org.apache.hadoop.hbase.CompareOperator".equals(keyValueVo.getKey())) {
                                CompareOperator compareOperator = CompareOperator.valueOf(keyValueVo.getValue());
                                objects.add(compareOperator);
                            }
                            if ("org.apache.hadoop.hbase.filter.ByteArrayComparable".equals(keyValueVo.getKey())) {
                                objects.add(new BinaryComparator(keyValueVo.getValue().getBytes()));

                            }
                            if ("org.apache.hadoop.hbase.filter.BitComparator.BitwiseOp".equals(keyValueVo.getKey())) {
                                objects.add(BitComparator.BitwiseOp.valueOf(keyValueVo.getValue()));

                            }

                        });
                        filterObj.set((Filter) constructor.newInstance(objects.toArray()));
                        filterList.addFilter(filterObj.get());
                        return true;
                    } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
                        throw new RRException("找不到这个查询拦截器构造器,检查这个json的字段顺序,要跟源码构造器字段顺序一致");
                    }

                }
                return false;
            }).findFirst();
        });

        return filterList;
    }

}
